package com.scala.MyUserActiveDegreeAnalyze

import org.apache.spark.sql.SparkSession

object MyScalaActiveAnalyze {
    case class UserActionLog(logId:Long,userId:Long,actionTime:String,actionType:Long,purchaseMoney:Double)
    case class UserActionLogValue(logId:Long,userId:Long,actionValue:Long)
    case class UserActionLogWithPurchaseMoneyValue(logId:Long,userId:Long,purchaseMoney:Double)

    def main(args: Array[String]): Unit = {

        val startDate = "2016-09-01";
        val endDate = "2016-11-01";

        val spark=SparkSession.builder().appName("UserAnalyze").master("local")
            //.config()
            .getOrCreate()
        //导入spark隐式转换
        import spark.implicits._
        //导入spark sql的functions
        import org.apache.spark.sql.functions._
        //获取数据集
        val userBaseInfo = spark.read.json("D:\\study\\拟定新\\user_base_info.json")
        val userActionLog=spark.read.json("D:\\study\\拟定新\\user_action_log.json")


    /*    //第一个功能：统计指定时间范围内访问次数最多10个用户

        //①过滤数据，找到指定时间范围内数据
        userActionLog .filter("actionTime >= '" + startDate + "' and actionTime <= '" + endDate + "' and actionType = 0")
        //②关联对应的用户基本信息数据
            .join(userBaseInfo, userActionLog("userId") === userBaseInfo("userId"))
            //③分组，按照userid和username
            .groupBy(userBaseInfo("userId"), userBaseInfo("username"))
            //④进行聚合
            .agg(count(userActionLog("logId")).alias("actionCount"))
            //⑤排序
            .sort($"actionCount".desc)
        //⑥抽取指定条数
            .limit(10)
            //⑦展示
            .show()*/
        //第二个功能：获取指定时间范围内购买金额最多的10个用户
  /*      userActionLog
            .filter("actionTime >= '" + startDate + "' and actionTime <= '" + endDate + "' and actionType = 1")
            .join(userBaseInfo, userActionLog("userId") === userBaseInfo("userId"))
            .groupBy(userBaseInfo("userId"), userBaseInfo("username"))
            .agg(round(sum(userActionLog("purchaseMoney")),2).alias("totalPurchaseMoney"))
            .sort($"totalPurchaseMoney".desc)
            .limit(10)
            .show()

        //第三个功能：统计最近一个周期相对上一个周期访问次数增长最多的10个用户
        *设定周期为一个月
          * 比如有一个用户，张三，张三9月份访问一共100次，10月份周期访问200次，
          * 这个总周期增长100次
          * 获取最近两个周期内，访问次数增长最多的10个用户
          *
          * 周期，自定义
          * 按一个月算 2016-10-01~2016-10-31，上一个周期就是2016-09-01~2016-09-30
        val userActionNewPeriod= userActionLog.as[UserActionLog]
            .filter("actionTime>='2016-10-01' and actionTime<='2016-10-31' and actionType=0")
            .map(userActionLogEntry=>UserActionLogValue(userActionLogEntry.logId,userActionLogEntry.userId,1))
        val userActionLastPeriod=userActionLog.as[UserActionLog]
            .filter("actionTime>='2016-09-01' and actionTime<='2016-09-30' and actionType=0")
            .map{userActionLogEntry=>UserActionLogValue(userActionLogEntry.logId,userActionLogEntry.userId,-1)}

        val userActionLogsDataSet = userActionNewPeriod.union(userActionLastPeriod)
        userActionLogsDataSet.join(userBaseInfo,userActionLogsDataSet("userId")===userBaseInfo("userId"))
            .groupBy(userBaseInfo("userId"),userBaseInfo("username"))
            .agg(sum(userActionLogsDataSet("actionValue")).alias(("actionIncr")))
            .sort($"actionIncr".desc)
            .limit(10)
            .show()

*/


        //第四个功能：最近周期内相对之前一个周期购买商品金额增长最快的10个用户
//         val userActionMoneyNewPeriod=userActionLog.as[UserActionLog]
//             .filter("actionTime>='2016-10-01' and actionTime<='2016-10-31' and actionType=1")
//             .map(actionEntry=>UserActionLogWithPurchaseMoneyValue(actionEntry.logId,actionEntry.userId,actionEntry.purchaseMoney))
//         val userActionMoneyLastPeriod=userActionLog.as[UserActionLog]
//             .filter("actionTime>='2016-09-01' and actionTime<='2016-09-30' and actionType=1")
//             .map(actionEntry=>UserActionLogWithPurchaseMoneyValue(actionEntry.logId,actionEntry.userId,-actionEntry.purchaseMoney))
//        val userActionPeriodMoneyDataSet = userActionMoneyNewPeriod.union(userActionMoneyLastPeriod)
//        userActionPeriodMoneyDataSet.join(userBaseInfo,userActionMoneyNewPeriod("userId")===userBaseInfo("userId"))
//            .groupBy(userBaseInfo("userId"),userBaseInfo("username"))
//            .agg(round(sum(userActionPeriodMoneyDataSet("purchaseMoney")),2).alias("purchaseMoneyIncr"))
//            .limit(10)
//            .show()

        //第五个功能：统计指定注册时间范围内头7天访问次数最高的10个用户
        userActionLog.join(userBaseInfo,userActionLog("userId")===userBaseInfo("userId"))
            .filter(userBaseInfo("registTime")>= "2016-10-01"
                && userBaseInfo("registTime")<="2016-10-31"
                && userActionLog("actionTime")>=userBaseInfo("registTime")
                && userActionLog("actionTime")<=date_add(userBaseInfo("registTime"),7)
                && userActionLog("actionType")===0
            ).groupBy(userBaseInfo("userId"),userBaseInfo("username"))
            .agg(count(userActionLog("logId")).alias("actionCount"))
            .sort($"actionCount".desc)
            .limit(10)
            .show()


        //第六个功能：统计指定注册时间范围内头7天购买金额最高的10个用户
        userActionLog.join(userBaseInfo,userActionLog("userId")===userBaseInfo("userId"))
            .filter(userBaseInfo("registTime")>="2016-10-01" && userBaseInfo("registTime")<="2016-10-31" && userActionLog("actionTime")>=userBaseInfo("registTime")&& userActionLog("actionTime")<=date_add(userBaseInfo("registTime"),7))
            .groupBy(userActionLog("userId"),userBaseInfo("username"))
            .agg(sum(userActionLog("purchaseMoney")).alias("MoneyBuy"))
            .sort($"MoneyBuy".desc)
            .limit(10)
            .show()

    }

}



















